package com.zp.qualitypractice.rx;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.internal.operators.flowable.FlowableOnBackpressureError;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 * Created by izp on 2017/8/29.
 * 原理：
 * 这个有问题吧，我看你注册的时候还分了时订阅PublishProcessor和PublishSubject二种，但是在post信息的时候，
 * 却是直接二个都直接调用onNext方法，那不是我PublishProcessor订阅了，PublishSubject也订阅了，我现在就想让PublishSubject发信息，
 * 结果让PublishProcessor也发送信息了。。。
    这里区分主要是在注册时候如果有背压的需求，就注册到PublishProcessor里面，如果没有就注册到PublishSubject中，
 这时候如果有消息，要两个都发信息的，因为不知道用户具体注册到哪个中去，所以两个都发，有注册的才会收到，
 当然你注册了两次，当然也会收到两次，注册一次就收到一次，这不是很正常吗
 */

public class ARxBus {
    private static ARxBus instance;

    private Subject<Object> subjectBus;

    private FlowableProcessor<Object> processorBus;

    private ARxBus() {
    }

    public static ARxBus getDefault() {
        if (instance == null) {
            synchronized (ARxBus.class) {
                if (instance == null) {
                    ARxBus tempInstance = new ARxBus();
                    tempInstance.subjectBus = PublishSubject.create().toSerialized();
                    tempInstance.processorBus = PublishProcessor.create().toSerialized();
                    instance = tempInstance;
                }
            }
        }
        return instance;
    }

    public Disposable register(Class eventType, Consumer observer) {
        return toObserverable(eventType).subscribe(observer);
    }

    public Disposable register(Class eventType, Consumer observer, Scheduler scheduler) {
        return toObserverable(eventType).observeOn(scheduler).subscribe(observer);
    }

    public Disposable register(Class eventType, Consumer observer,Scheduler scheduler, BackpressureStrategy strategy){
        Flowable o = toFlowable(eventType);
        switch (strategy) {
            case DROP:
                o = o.onBackpressureDrop();
            case LATEST:
                o = o.onBackpressureLatest();
            case MISSING:
                o = o;
            case ERROR:
                o = RxJavaPlugins.onAssembly(new FlowableOnBackpressureError<>(o));
            default:
                o = o.onBackpressureBuffer();
        }
        if(scheduler!=null){
            o.observeOn(scheduler);
        }
        return o.subscribe(observer);
    }

    public Disposable register(Class eventType, Consumer observer,BackpressureStrategy strategy){
        return register(eventType,observer,null,strategy);
    }

    public void unRegister(Disposable disposable) {
        if (disposable != null && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }

    public void unRegister(CompositeDisposable compositeDisposable) {
        if (compositeDisposable != null) {
            compositeDisposable.dispose();
        }
    }

    public void post(final Object event) {
        subjectBus.onNext(event);
        processorBus.onNext(event);
    }

    private Observable toObserverable(Class cls) {
        return subjectBus.ofType(cls);
    }

    private Flowable toFlowable(Class cls) {
        return processorBus.ofType(cls);
    }

    public boolean hasObservers() {
        return subjectBus.hasObservers();
    }

    public boolean hasSubscribers() {
        return processorBus.hasSubscribers();
    }

}
